package cn.edu.zut.qixiang.service;

import cn.edu.zut.qixiang.config.RealTimeWeatherConfig;
import cn.edu.zut.qixiang.entity.WeatherData;
import cn.edu.zut.qixiang.repository.WeatherDataRepository;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 *
 * @author 86178
 */
@Service
@Slf4j
public class RealTimeWeatherProducerService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private RealTimeWeatherConfig config;

    @Autowired
    private WeatherDataRepository weatherDataRepository;
    // 存储最近数据用于前端展示 - 使用LinkedHashMap保持插入顺序
    private final Map<String, LinkedHashMap<String, Map<String, Object>>> recentData = new ConcurrentHashMap<>();
    private final AtomicLong messageCount = new AtomicLong(0);
    private final AtomicLong errorCount = new AtomicLong(0);
    private LocalDateTime serviceStartTime;

    private final Random random = new Random();
    private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm");
    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private volatile boolean kafkaAvailable = true;
    private volatile boolean isRunning = true;

    // 缓存基础数据，避免重复计算
    private final Map<String, Double> districtBaseTemperatures = new ConcurrentHashMap<>();
    private final Map<String, Double> districtBaseWindSpeeds = new ConcurrentHashMap<>();

    public RealTimeWeatherProducerService() {
        this.serviceStartTime = LocalDateTime.now();
    }

    /**
     * 使用 @PostConstruct 确保在依赖注入完成后初始化
     */
    @PostConstruct
    public void init() {
        log.info("开始初始化实时气象数据服务...");
        if (config == null || config.getDistricts() == null) {
            log.error("配置未正确初始化，无法启动服务");
            return;
        }

        initializeDistrictBaseData();
        initializeRecentData();

        // 添加MySQL连接测试
        try {
            long count = weatherDataRepository.count();
            log.info("MySQL连接成功，当前记录数: {}", count);
        } catch (Exception e) {
            log.error("MySQL连接失败: {}", e.getMessage());
            log.warn("将继续运行，但数据不会保存到MySQL");
        }

        log.info("实时气象数据服务初始化完成，支持区域: {}个", config.getDistricts().size());
    }

    /**
     * 初始化区域基础数据缓存
     */
    private void initializeDistrictBaseData() {
        // 初始化基础温度
        Map<String, Double> baseTemps = new HashMap<>();
        baseTemps.put("中原区", 25.6);
        baseTemps.put("二七区", 26.1);
        baseTemps.put("管城回族区", 25.8);
        baseTemps.put("金水区", 26.3);
        baseTemps.put("上街区", 24.9);
        baseTemps.put("惠济区", 25.2);
        baseTemps.put("中牟县", 25.1);
        baseTemps.put("巩义市", 24.8);
        baseTemps.put("荥阳市", 25.4);
        baseTemps.put("新密市", 25.0);
        baseTemps.put("新郑市", 25.7);
        baseTemps.put("登封市", 24.6);
        districtBaseTemperatures.putAll(baseTemps);

        // 初始化基础风速
        int index = 0;
        for (String district : config.getDistricts()) {
            districtBaseWindSpeeds.put(district, 2.5 + (index % 3) * 0.3);
            index++;
        }
    }

    /**
     * 初始化最近数据存储
     */
    private void initializeRecentData() {
        LocalDateTime now = LocalDateTime.now();

        for (String district : config.getDistricts()) {
            LinkedHashMap<String, Map<String, Object>> districtData = new LinkedHashMap<String, Map<String, Object>>() {
                @Override
                protected boolean removeEldestEntry(Map.Entry<String, Map<String, Object>> eldest) {
                    return size() > config.getDataRetentionMinutes();
                }
            };

            // 预生成初始数据
            for (int i = config.getDataRetentionMinutes() - 1; i >= 0; i--) {
                LocalDateTime timestamp = now.minusMinutes(i);
                Map<String, Object> data = generateDistrictWeatherData(district, timestamp);
                String timeKey = timestamp.format(TIME_FORMATTER);
                districtData.put(timeKey, data);
            }

            recentData.put(district, districtData);
        }
    }
    /**
     * 保存数据到MySQL
     */
    private void saveToMySQL(Map<String, Object> weatherData) {
        try {
            WeatherData entity = convertToEntity(weatherData);
            WeatherData saved = weatherDataRepository.save(entity);

            if (log.isDebugEnabled()) {
                log.debug("数据已保存到MySQL，ID: {} - {}", saved.getId(), saved.getDistrict());
            }
        } catch (Exception e) {
            log.error("保存数据到MySQL失败: {} - {}", weatherData.get("district"), e.getMessage());
            // 不抛出异常，避免影响主流程
        }
    }
    /**
     * 将Map数据转换为Entity
     */
    private WeatherData convertToEntity(Map<String, Object> weatherData) {
        WeatherData entity = new WeatherData();
        entity.setDistrict((String) weatherData.get("district"));
        entity.setTimestamp(LocalDateTime.parse(
                (String) weatherData.get("timestamp"),
                DATE_TIME_FORMATTER
        ));
        entity.setTemperature((Double) weatherData.get("temperature"));
        entity.setHumidity((Integer) weatherData.get("humidity"));
        entity.setWindSpeed((Double) weatherData.get("windSpeed"));
        entity.setWindDirection((String) weatherData.get("windDirection"));
        entity.setPressure((Double) weatherData.get("pressure"));
        entity.setPrecipitation((Double) weatherData.get("precipitation"));
        entity.setCreateTime(LocalDateTime.now());

        return entity;
    }

    /**
     * 每分钟生成实时数据
     */
    @Scheduled(fixedRateString = "${realtime.weather.update-interval-ms:60000}")
    public void generateRealTimeData() {
        if (!isRunning || config == null || config.getDistricts() == null) {
            return;
        }

        LocalDateTime currentTime = LocalDateTime.now();
        String currentTimeKey = currentTime.format(TIME_FORMATTER);

        if (log.isDebugEnabled()) {
            log.debug("生成实时气象数据 - 时间: {}", currentTime.format(DATE_TIME_FORMATTER));
        }

        // 使用并行流提高生成效率
        List<Map<String, Object>> weatherDataList = config.getDistricts().parallelStream()
                .map(district -> {
                    Map<String, Object> weatherData = generateDistrictWeatherData(district, currentTime);
                    updateRecentData(district, currentTimeKey, weatherData);
                    saveToMySQL(weatherData);
                    return weatherData;
                })
                .collect(Collectors.toList());

        // 批量发送到Kafka
        batchSendToKafka(weatherDataList);

        if (log.isDebugEnabled()) {
            log.debug("实时数据生成完成，总消息数: {}", messageCount.get());
        }
    }

    /**
     * 批量发送数据到Kafka
     */
    private void batchSendToKafka(List<Map<String, Object>> weatherDataList) {
        if (!kafkaAvailable || weatherDataList.isEmpty()) {
            return;
        }

        List<ListenableFuture<SendResult<String, Object>>> futures = new ArrayList<>();

        for (Map<String, Object> weatherData : weatherDataList) {
            try {
                ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(
                        config.getTopic(),
                        (String) weatherData.get("district"), // 使用地区作为key，确保相同地区的数据在同一个分区
                        weatherData
                );

                future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                    @Override
                    public void onSuccess(SendResult<String, Object> result) {
                        messageCount.incrementAndGet();
                        if (log.isTraceEnabled()) {
                            log.trace("数据发送成功: {} - {}",
                                    weatherData.get("district"), weatherData.get("timeDisplay"));
                        }
                    }

                    @Override
                    public void onFailure(Throwable ex) {
                        errorCount.incrementAndGet();
                        log.error("发送Kafka消息失败: {} - {}",
                                weatherData.get("district"), ex.getMessage());
                        kafkaAvailable = false;
                    }
                });

                futures.add(future);
            } catch (Exception e) {
                errorCount.incrementAndGet();
                log.error("发送Kafka消息异常: {} - {}",
                        weatherData.get("district"), e.getMessage());
                kafkaAvailable = false;
            }
        }

    }

    /**
     * 生成单个区域的实时气象数据
     */
    private Map<String, Object> generateDistrictWeatherData(String district, LocalDateTime timestamp) {
        Map<String, Object> data = new HashMap<>(16); // 预设初始容量

        // 基础信息
        data.put("district", district);
        data.put("timestamp", timestamp.format(DATE_TIME_FORMATTER));
        data.put("timeDisplay", timestamp.format(TIME_FORMATTER));
        data.put("dataId", System.currentTimeMillis() + random.nextInt(1000));

        // 生成四个图表需要的数据
        generateWindData(data, district, timestamp);
        generatePressureData(data, district, timestamp);
        generateTemperatureData(data, district, timestamp);
        generatePrecipitationData(data, district, timestamp);

        return data;
    }

    /**
     * 生成风况数据
     */
    private void generateWindData(Map<String, Object> data, String district, LocalDateTime timestamp) {
        int hour = timestamp.getHour();
        int minute = timestamp.getMinute();

        // 使用缓存的基础风速
        double baseWindSpeed = districtBaseWindSpeeds.getOrDefault(district, 2.5);
        double timeFactor = Math.sin((hour + minute / 60.0) * Math.PI / 12) * 0.5;
        double randomFactor = (random.nextDouble() - 0.5) * 1.0;

        double windSpeed = Math.max(0.1, Math.min(baseWindSpeed + timeFactor + randomFactor, 15.0));
        data.put("windSpeed", round(windSpeed, 1));
        data.put("windDirection", generateWindDirection(hour));
    }

    /**
     * 生成气压数据
     */
    private void generatePressureData(Map<String, Object> data, String district, LocalDateTime timestamp) {
        int hour = timestamp.getHour();

        // 预计算的值
        double diurnalVariation = Math.sin((hour - 10) * Math.PI / 12) * 2.0;
        double districtVariation = (getDistrictIndex(district) - 6) * 0.3;
        double randomFluctuation = (random.nextDouble() - 0.5) * 1.5;

        double pressure = 1013.0 + diurnalVariation + districtVariation + randomFluctuation;
        data.put("pressure", round(pressure, 1));
    }

    /**
     * 生成温度数据
     */
    private void generateTemperatureData(Map<String, Object> data, String district, LocalDateTime timestamp) {
        int hour = timestamp.getHour();
        int month = timestamp.getMonthValue();

        // 使用缓存的基础温度
        double baseTemp = districtBaseTemperatures.getOrDefault(district, 25.0);
        double diurnalVariation = calculateTemperatureDiurnalVariation(hour, month);
        double urbanHeatEffect = isUrbanDistrict(district) ? 1.5 : -0.5;
        double randomFluctuation = (random.nextDouble() - 0.5) * 1.0;

        double temperature = baseTemp + diurnalVariation + urbanHeatEffect + randomFluctuation;
        data.put("temperature", round(temperature, 1));
        data.put("humidity", generateHumidity(hour));
    }

    /**
     * 生成降水数据
     */
    private void generatePrecipitationData(Map<String, Object> data, String district, LocalDateTime timestamp) {
        int hour = timestamp.getHour();
        int month = timestamp.getMonthValue();

        double precipitationProbability = getPrecipitationProbability(month, hour);
        double precipitation = random.nextDouble() < precipitationProbability ?
                random.nextDouble() * 3.0 : 0.0;

        data.put("precipitation", round(precipitation, 2));
    }

    /**
     * 更新最近数据
     */
    private void updateRecentData(String district, String timeKey, Map<String, Object> newData) {
        LinkedHashMap<String, Map<String, Object>> districtData = recentData.get(district);
        if (districtData != null) {
            districtData.put(timeKey, newData);
        }
    }

    /**
     * 尝试恢复Kafka连接 - 增加重试机制
     */
    @Scheduled(fixedRate = 30000)
    public void tryRecoverKafka() {
        if (!kafkaAvailable) {
            log.info("尝试恢复Kafka连接...");
            try {
                Map<String, Object> testMsg = new HashMap<>();
                testMsg.put("test", true);
                testMsg.put("timestamp", LocalDateTime.now().format(DATE_TIME_FORMATTER));
                testMsg.put("service", "RealTimeWeatherProducerService");

                ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(
                        config.getTopic(), "health-check", testMsg
                );

                future.addCallback(
                        result -> {
                            kafkaAvailable = true;
                            log.info("Kafka连接已恢复");
                        },
                        ex -> {
                            log.warn("Kafka连接恢复失败: {}", ex.getMessage());
                        }
                );
            } catch (Exception e) {
                log.warn("Kafka连接恢复异常: {}", e.getMessage());
            }
        }
    }
    private final ObjectMapper objectMapper = new ObjectMapper();
    private volatile boolean flinkDataReceived = false;
    private final AtomicLong flinkMessageCount = new AtomicLong(0);
    @KafkaListener(topics = "flink_processed_weather_data", groupId = "frontend-consumer")
    public void consumeFlinkProcessedData(String message) {
        try {
            log.info("收到Flink处理数据: {}", message);

            // 手动解析JSON
            JsonNode jsonNode = objectMapper.readTree(message);

            // 安全地提取数据字段
            String district = jsonNode.has("district") ? jsonNode.get("district").asText() : "未知区域";
            double temperature = jsonNode.has("temperature") ? jsonNode.get("temperature").asDouble() : 0.0;
            double windSpeed = jsonNode.has("windSpeed") ? jsonNode.get("windSpeed").asDouble() : 0.0;
            double pressure = jsonNode.has("pressure") ? jsonNode.get("pressure").asDouble() : 0.0;
            double precipitation = jsonNode.has("precipitation") ? jsonNode.get("precipitation").asDouble() : 0.0;
            int humidity = jsonNode.has("humidity") ? jsonNode.get("humidity").asInt() : 0;

            String timeKey = LocalDateTime.now().format(TIME_FORMATTER);

            Map<String, Object> weatherData = new HashMap<>();
            weatherData.put("district", district);
            weatherData.put("timestamp", LocalDateTime.now().format(DATE_TIME_FORMATTER));
            weatherData.put("timeDisplay", timeKey);
            weatherData.put("temperature", temperature);
            weatherData.put("humidity", humidity);
            weatherData.put("windSpeed", windSpeed);
            weatherData.put("windDirection", "东风");
            weatherData.put("pressure", pressure);
            weatherData.put("precipitation", precipitation);

            updateRecentData(district, timeKey, weatherData);

            flinkDataReceived = true;
            flinkMessageCount.incrementAndGet();

            log.debug("成功处理Flink数据: {} - 温度: {}°C", district, temperature);

        } catch (Exception e) {
            log.error("处理Flink转发数据失败: {}, 原始消息: {}", e.getMessage(), message);
        }
    }

    /**
     * 获取指定区域的最近气象数据
     */
    public Map<String, Object> getRecentWeatherData(String district) {
        Map<String, Object> result = new HashMap<>();

        if (!isFlinkJobRunning()) {
            result.put("error", "Flink作业未运行，无法获取实时数据");
            result.put("source", "flink_stopped");
            return result;
        }

        LinkedHashMap<String, Map<String, Object>> districtData = recentData.get(district);
        if (districtData != null && !districtData.isEmpty()) {

            List<Map<String, Object>> dataList = new ArrayList<>(districtData.values());
            result.put("district", district);
            result.put("data", dataList);
            result.put("lastUpdate", LocalDateTime.now().format(DATE_TIME_FORMATTER));
            result.put("dataCount", dataList.size());
        } else {
            result.put("error", "区域数据未初始化: " + district);
        }

        return result;
    }

    /**
     * 获取所有区域的当前数据快照 - 优化性能
     */
    public Map<String, Object> getAllDistrictsCurrentData() {
        Map<String, Object> result = new HashMap<>();
        Map<String, Object> currentData = new HashMap<>();

        if (config == null || config.getDistricts() == null) {
            result.put("error", "服务未就绪");
            return result;
        }

        config.getDistricts().forEach(district -> {
            LinkedHashMap<String, Map<String, Object>> districtData = recentData.get(district);
            if (districtData != null && !districtData.isEmpty()) {
                // 获取最新的数据（LinkedHashMap的最后一条）
                Map<String, Object> latestData = getLatestData(districtData);
                if (latestData != null) {
                    currentData.put(district, latestData);
                }
            }
        });

        result.put("currentData", currentData);
        result.put("timestamp", LocalDateTime.now().format(DATE_TIME_FORMATTER));
        result.put("totalDistricts", config.getDistricts().size());

        return result;
    }

    /**
     * 获取服务状态 - 增加更多监控指标
     */
    public Map<String, Object> getServiceStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("isRunning", isRunning);
        status.put("kafkaAvailable", kafkaAvailable);
        status.put("serviceStartTime", serviceStartTime.format(DATE_TIME_FORMATTER));
        status.put("uptimeMinutes", java.time.Duration.between(serviceStartTime, LocalDateTime.now()).toMinutes());
        status.put("totalMessagesSent", messageCount.get());
        status.put("totalErrors", errorCount.get());
        status.put("successRate", calculateSuccessRate());

        if (config != null) {
            status.put("supportedDistricts", config.getDistricts());
            status.put("kafkaTopic", config.getTopic());
            status.put("dataRetentionMinutes", config.getDataRetentionMinutes());
            status.put("updateIntervalMs", config.getUpdateIntervalMs());

            // 统计每个区域的数据量
            Map<String, Integer> dataCounts = new HashMap<>();
            config.getDistricts().forEach(district -> {
                LinkedHashMap<String, Map<String, Object>> data = recentData.get(district);
                dataCounts.put(district, data != null ? data.size() : 0);
            });
            status.put("dataCounts", dataCounts);
        }

        return status;
    }

    /**
     * 停止服务
     */
    public void stopService() {
        isRunning = false;
        log.info("实时气象数据服务已停止");
    }

    /**
     * 启动服务
     */
    public void startService() {
        isRunning = true;
        log.info("实时气象数据服务已启动");
    }

    // ========== 辅助方法 ==========

    // 检查Flink作业状态
    public boolean isFlinkJobRunning() {
        return flinkDataReceived && flinkMessageCount.get() > 0;
    }

    private Map<String, Object> getLatestData(LinkedHashMap<String, Map<String, Object>> districtData) {
        if (districtData.isEmpty()) {
            return null;
        }
        // 获取最后一个entry的值（最新的数据）
        return districtData.entrySet().stream()
                .reduce((first, second) -> second)
                .map(Map.Entry::getValue)
                .orElse(null);
    }

    private double calculateSuccessRate() {
        long total = messageCount.get() + errorCount.get();
        return total > 0 ? (double) messageCount.get() / total * 100 : 100.0;
    }

    private String generateWindDirection(int hour) {
        String[] directions = {"北", "东北", "东", "东南", "南", "西南", "西", "西北"};
        int baseIndex = (hour / 3) % directions.length;
        int variation = random.nextInt(3) - 1;
        return directions[(baseIndex + variation + directions.length) % directions.length];
    }

    private double calculateTemperatureDiurnalVariation(int hour, int month) {
        double amplitude = getDiurnalAmplitude(month);
        double minTempHour = 5.5;
        double maxTempHour = 14.5;

        double hourDiff = (hour - minTempHour + 24) % 24; // 处理跨天情况
        return hourDiff <= (maxTempHour - minTempHour) ?
                amplitude * (hourDiff / (maxTempHour - minTempHour)) :
                amplitude * (1 - (hourDiff - (maxTempHour - minTempHour)) / (24 - (maxTempHour - minTempHour)));
    }

    private double getDiurnalAmplitude(int month) {
        switch (month) {
            case 5:
                return 10.0;
            case 6:
                return 12.0;
            case 7:
                return 13.0;
            case 8:
                return 11.0;
            case 9:
                return 9.0;
            default:
                return 10.0;
        }
    }

    private int generateHumidity(int hour) {
        double baseHumidity = 65 - 25 * Math.sin(2 * Math.PI * (hour - 4) / 24.0);
        double randomNoise = (random.nextDouble() - 0.5) * 15;
        return (int) Math.max(30, Math.min(95, baseHumidity + randomNoise));
    }

    private double getPrecipitationProbability(int month, int hour) {
        double baseProbability;
        switch (month) {
            case 5:
                baseProbability = 0.05;
                break;
            case 6:
                baseProbability = 0.08;
                break;
            case 7:
                baseProbability = 0.12;
                break;
            case 8:
                baseProbability = 0.10;
                break;
            case 9:
                baseProbability = 0.06;
                break;
            default:
                baseProbability = 0.11;
                break;
        }

        return (hour >= 14 && hour <= 20) ? baseProbability * 1.5 : baseProbability;
    }

    private boolean isUrbanDistrict(String district) {
        return district.contains("区") && !district.equals("上街区");
    }

    private int getDistrictIndex(String district) {
        List<String> districts = config.getDistricts();
        for (int i = 0; i < districts.size(); i++) {
            if (districts.get(i).equals(district)) {
                return i;
            }
        }
        return 0;
    }

    private double round(double value, int places) {
        double scale = Math.pow(10, places);
        return Math.round(value * scale) / scale;
    }
}